package com.dux.commons.network;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import java.nio.channels.SelectableChannel;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.nio.channels.spi.SelectorProvider;

import org.apache.log4j.Logger;

/**
 * 调度通过选择器选择调度选择键集
 * @author 杜祥
 * @create 2013-6-1
 */
public abstract class Dispatcher extends Thread
{
	
	private static Logger log = Logger.getLogger(Dispatcher.class);
	
	/**
	 * 选择器选择准备的键
	 */
	Selector selector;
	
	/**
	 * 线程池，用于执行断开连接的任务
	 */
	private final DisconnectionThreadPool dcPool;
	
	/**
	 * 用于注册与selector.select同步的对象
	 */
	private final Object gate = new Object();
	
	
	/**
	 * 构造一个调度器，并设置名字和断开连接任务执行的线程池
	 * @param name
	 * @param dcPool
	 * @throws IOException
	 */
	public Dispatcher(String name, DisconnectionThreadPool dcPool) throws IOException
	{
		super(name);
		this.selector = SelectorProvider.provider().openSelector();
		this.dcPool = dcPool;
	}
	
	/**
	 * 添加连接到关闭列表，以便调度器尽快关闭这个连接。
	 * @param connection
	 */
	abstract void closeConnection(MMOConnection connection);
	
	
	
	/**
	 * 调度选择键并处理等待关闭。
	 * @throws IOException
	 */
	abstract void dispatch() throws IOException;
	
	
	/**
	 * 获得选择器
	 * @return			选择器
	 */
	public final Selector selector()
	{
		return this.selector;
	}

	
	/**
	 * 
	 * 
	 * @see java.lang.Thread#run()
	 */
	@Override
	public void run() 
	{
		for(;;)
		{
			try
			{
				dispatch();
				
				synchronized (gate) 
				{
					
				}
			} 
			catch (IOException e) 
			{
				e.printStackTrace();
			}
		}
	}
	
	
	/**
	 * 注册新的客户端连接到这个调度程序,并设置客户端连接的SelectionKey。
	 * @param ch
	 * @param ops
	 * @param connection
	 * @throws IOException
	 */
	public final void register(SelectableChannel ch, int ops, MMOConnection connection) throws IOException
	{
		synchronized(gate)
		{
			selector.wakeup();
			connection.setKey(ch.register(selector, ops, connection));
		}
	}
	
	
	/**
	 * 注册一个新的接受器到这个调度程序并返回一个SelectionKey.
	 * @param ch
	 * @param ops
	 * @param acceptor
	 * @return	
	 * @throws IOException
	 */
	public final SelectionKey register(SelectableChannel ch, int ops, Acceptor acceptor) throws IOException
	{
		synchronized(gate)
		{
			selector.wakeup();
			return ch.register(selector, ops, acceptor);
		}
	}
	
	
	/**
	 * 接收
	 * @param key
	 */
	final void accept(SelectionKey key)
	{
		try
		{
			((Acceptor)key.attachment()).accept(key);
		}
		catch (Exception e) 
		{
			log.error("接受连接时出错！", e);
		}
	}
	
	
	/**
	 * 读取数据
	 */
	final void read(SelectionKey key)
	{
		SocketChannel channel = (SocketChannel)key.channel();
		MMOConnection connection = (MMOConnection)key.attachment();
		
		ByteBuffer buffer = connection.readBuffer;
		
		int numRead;
		try 
		{
			numRead = channel.read(buffer);
		} 
		catch (IOException e)
		{
			closeConnectionImpl(connection);
			return;
		}
		/*通道是否达到末尾*/
		if(numRead == -1)
		{
			closeConnectionImpl(connection);
			return ;
		}
		/*没有数据*/
		else if(numRead == 0)
		{
			return ;
		}
		buffer.flip();
		/*分包*/
		while(buffer.remaining() > 4 && buffer.remaining() >= buffer.getInt(buffer.position()))
		{
			if(!parse(connection, buffer))
			{
				closeConnectionImpl(connection);
				return;
			}
		}
		
		if(buffer.hasRemaining())
		{
			buffer.compact();
		}
		else
		{
			buffer.clear();
		}
	}
	
	/**
	 * 从连接中读取一个数据包，并执行。
	 * @param connection
	 * @param buffer
	 * @return
	 */
	private boolean parse(MMOConnection connection, ByteBuffer buffer)
	{
		int length = 0;
		try 
		{
			length = buffer.getInt();
			
			ByteBuffer _buffer = (ByteBuffer) buffer.slice().limit(length);
			
			_buffer.order(ByteOrder.LITTLE_ENDIAN);
			/*设置缓冲区的位置*/
			buffer.position(buffer.position() + length);
			
			return connection.processData(_buffer);
		} 
		catch (IllegalArgumentException e)
		{
			log.warn("解析客户端的包出错-连接" + connection + "包长:" + length + "remaining:" + buffer.remaining(), e);
		}
		return false;
	}
	
	
	/**
	 * 写入数据
	 */
	final void write(SelectionKey key)
	{
		SocketChannel channel = (SocketChannel)key.channel();
		MMOConnection connection = (MMOConnection)key.attachment();
		
		ByteBuffer buffer = connection.writeBuffer;
		int numWrite;
		if(buffer.hasRemaining())
		{
			try 
			{
				numWrite = channel.write(buffer);
			} 
			catch (IOException e)
			{
				closeConnectionImpl(connection);
				return ;
			}
			if(numWrite == 0)
			{
				log.info("无法写入 : " + connection.getIp());
				return;
			}
			/*所有数据无法全部写入*/
			if(buffer.hasRemaining())
				return;
		}
		
		while(true)
		{
			buffer.clear();
			boolean writeFaild = connection.writeData(buffer);
			if(!writeFaild)
			{
				buffer.limit(0);
				break;
			}
			
			try 
			{
				numWrite = channel.write(buffer);
			} 
			catch (IOException e)
			{
				closeConnectionImpl(connection);
				return ;
			}
			
			if(numWrite == 0)
			{
				log.info("无法写入 : " + connection.getIp());
				return;
			}
			/*所有数据无法全部写入*/
			if(buffer.hasRemaining())
				return;
		}
		/*已经写完所有的数据，所以不在此通道写入数据*/
		key.interestOps(key.interestOps() & ~SelectionKey.OP_WRITE);
		
		/*发送最后一个包*/
		if(connection.isPendingClose())
			closeConnectionImpl(connection);
	}
	
	
	/**
	 * 关闭连接方法。<br>
	 * 此方法会执行onlyClose()和onDisconnect()方法
	 * @param con
	 */
	protected final void closeConnectionImpl(MMOConnection con)
	{
		if(con.onlyClose())
			dcPool.scheduleDisconnection(new DisconnectionTask(con), con.getDisconnectionDelay());
	}
}
